# split out from select_many
function cross_join(source::Query.Enumerable, inner_collection::Query.Enumerable, f_resultSelector::Function, resultSelector::Expr)
    TS = eltype(source)
    f_collectionSelector = i -> inner_collection

    input_type_collection_selector = typeof(inner_collection)
    TCE = input_type_collection_selector.parameters[1]

    T = Base.return_types(f_resultSelector, (TS,TCE))[1]
    SO = typeof(source)

    CS = typeof(f_collectionSelector)
    RS = typeof(f_resultSelector)

    Query.EnumerableSelectMany{T,SO,CS,RS}(source,f_collectionSelector,f_resultSelector)
end

function query_split_rows(source::Query.Enumerable, f_collectionSelector::Function, collectionSelector::Expr, f_resultSelector::Function, resultSelector::Expr)
    TS = eltype(source)

    input_type_collection_selector = Base.return_types(f_collectionSelector, (TS,))[1]
    TCE = typeof(input_type_collection_selector)==Union || input_type_collection_selector==Any ? Any : input_type_collection_selector.parameters[1]

    T = Base.return_types(f_resultSelector, (TS,TCE))[1]
    SO = typeof(source)

    CS = typeof(f_collectionSelector)
    RS = typeof(f_resultSelector)

    Query.EnumerableSelectMany{T,SO,CS,RS}(source,f_collectionSelector,f_resultSelector)
end

const AbstractQuery = Union{Query.Enumerable, Query.Queryable}

query_names(q::AbstractQuery) = fieldnames(eltype(q))

double_with(with_context, fields) = begin
    anon = WithContext(with_context,
        environment = merge!(get_environment(with_context), fields)
    ) |> my_macroexpand |> LazyContext.parse

    display = WithContext(with_context,
        environment = merge!(Environment(), fields)
    ) |>  LazyContext.parse |> Meta.quot

    anon, display
end

with_anonymous(with_context, field_names = Symbol[]; source_name = gensym()) = begin
    double_with(
        WithContext(with_context, expression = :($source_name -> $(get_expression(with_context)))),
        ImmutableDict(add_dots(field_names, source_name), :row => source_name))
end

"""
    @rows_from q::AbstractQuery *wherever* condition

Filter rows

```jldoctest
julia> using LazyQuery

julia> @new_environment;

julia> @chain @evaluate begin
            using LazyQuery
            DataFrame(
                name = ["John", "Sally", "Kirk"],
                age = [23, 42, 59],
                children = [3,5,2] )
            query(it)
            @rows_from it wherever age > 30 && children > 2
            @output it into DataFrame
        end
1×3 DataFrames.DataFrame
│ Row │ name    │ age │ children │
├─────┼─────────┼─────┼──────────┤
│ 1   │ "Sally" │ 42  │ 5        │
```
"""
rows_from(q::AbstractQuery, conjunction, which) =
    eval(get_module(which), :($(Query.where)($q, $(with_anonymous(which, query_names(q))...))))

"""
    @group q::AbstractQuery *by* column

Grouped vectors will come in as generators. You might need to collect them
before use.

```jldoctest
julia> using LazyQuery

julia> @new_environment;

julia> @chain @evaluate begin
            using LazyQuery
            DataFrame(
                a = [1, 1, 2, 2],
                b = [1, 2, 3, 4],
                c = [4, 3, 2, 1]
            )
            query(it)
            @group it by a
            @make_from it a b = sum(b) c = mean(collect(c))
            @output it into DataFrame
        end
2×3 DataFrames.DataFrame
│ Row │ a │ b │ c   │
├─────┼───┼───┼─────┤
│ 1   │ 1 │ 3 │ 3.5 │
│ 2   │ 2 │ 7 │ 1.5 │
```
"""
group(q::AbstractQuery, conjunction, column) = begin
    q_names = query_names(q)
    column_variable = get_expression(column)
    others = setdiff(q_names, [column_variable])
    result = eval(get_module(column), :($(Query.group_by)($q, $(with_anonymous(column, query_names(q))...))))
    make_from(result, map(others) do name
        :($name = $(Base.Generator)(i -> i.$name, row))
    end..., :($column_variable = key))
end

then_by(q::AbstractQuery, values) =
    MacroTools.@match get_expression(values) begin
        descending(order_) =>
            eval(get_module(values), :($(Query.thenby_descending)($q,
                $(with_anonymous(WithContext(values, expression = order), query_names(q))...))))
        any_ =>
            eval(get_module(values), :($(Query.thenby)($q, $(with_anonymous(values, query_names(q))...))))
        end

"""
    @order q::AbstractQuery *by* args...

You can use `descending` to reverse the order.

```jldoctest
julia> using LazyQuery

julia> @new_environment;

julia> @chain @evaluate begin
            using LazyQuery
            DataFrame(
                a = [3, 3, 2, 2, 1, 1],
                b = [1, 2, 2, 1, 1, 2])
            query(it)
            @order it by a descending(b)
            @output it into DataFrame
        end
6×2 DataFrames.DataFrame
│ Row │ a │ b │
├─────┼───┼───┤
│ 1   │ 1 │ 2 │
│ 2   │ 1 │ 1 │
│ 3   │ 2 │ 2 │
│ 4   │ 2 │ 1 │
│ 5   │ 3 │ 2 │
│ 6   │ 3 │ 1 │

julia> @chain @evaluate begin
            DataFrame(
                a = [3, 3, 2, 2, 1, 1],
                b = [1, 2, 2, 1, 1, 2])
            query(it)
            @order it by descending(a) b
            @output it into DataFrame
        end
6×2 DataFrames.DataFrame
│ Row │ a │ b │
├─────┼───┼───┤
│ 1   │ 3 │ 1 │
│ 2   │ 3 │ 2 │
│ 3   │ 2 │ 1 │
│ 4   │ 2 │ 2 │
│ 5   │ 1 │ 1 │
│ 6   │ 1 │ 2 │
```
"""
order(q::AbstractQuery, conjunction, values, args...) = begin
    result = MacroTools.@match get_expression(values) begin
        descending(ordercolumn_) =>
            eval(get_module(values), :($(Query.orderby_descending)($q,
                $(with_anonymous(WithContext(values, expression = ordercolumn), query_names(q))...))))
        any_ =>
            eval(get_module(values), :($(Query.orderby)($q, $(with_anonymous(values, query_names(q))...))))
        end
    for value in args
        result = then_by(result, value)
    end
    result
end

make_selection(selections...) =
    :($NamedTuples.@NT $(make_kw.(selections)...))

basic_select(q::AbstractQuery, first_selection, other_selections...; get = false) = begin
    first_expression = get_expression(first_selection)
    q_names = query_names(q)
    selection_expression =
        if get
            first_expression
        else
            make_selection(first_selection, other_selections...)
        end
    eval(get_module(first_selection), :($(Query.select)($q, $(with_anonymous(
        WithContext(first_selection, expression = selection_expression),
        q_names
    )...))))
end

"""
    @make_from q::AbstractQuery selections...

```jldoctest
julia> using LazyQuery

julia> @new_environment;

julia> @chain @evaluate begin
            using LazyQuery
            DataFrame(
                name = ["John", "Sally", "Kirk"],
                age = [23, 42, 59],
                children = [3,5,2] )
            query(it)
            @make_from it age name = lowercase(name)
            @output it into DataFrame
        end
3×2 DataFrames.DataFrame
│ Row │ age │ name    │
├─────┼─────┼─────────┤
│ 1   │ 23  │ "john"  │
│ 2   │ 42  │ "sally" │
│ 3   │ 59  │ "kirk"  │
```
"""
make_from(q::AbstractQuery, selections...) = basic_select(q, selections...)

"""
    @with q::AbstractQuery selection

```jldoctest
julia> using LazyQuery

julia> @new_environment;

julia> @chain @evaluate begin
            using LazyQuery
            DataFrame(
                name = ["John", "Sally", "Kirk"],
                age = [23, 42, 59],
                children = [3,5,2] )
            query(it)
            @with it lowercase(name)
            @output it
        end
3-element Array{DataValues.DataValue{String},1}:
 "john"
 "sally"
 "kirk"
```
"""
with(q::AbstractQuery, selection) =
    basic_select(q, selection, get = true)

"""
    @add_to q::AbstractQuery selections...

```jldoctest
julia> using LazyQuery

julia> @new_environment;

julia> @chain @evaluate begin
            using LazyQuery
            DataFrame(
                a = [1, 2, 3],
                b = [2, 3, 4] )
            query(it)
            @add_to it c = a + b d = a - b
            @add_to it e = c / d
            @output it into DataFrame
        end
3×5 DataFrames.DataFrame
│ Row │ a │ b │ c │ d  │ e    │
├─────┼───┼───┼───┼────┼──────┤
│ 1   │ 1 │ 2 │ 3 │ -1 │ -3.0 │
│ 2   │ 2 │ 3 │ 5 │ -1 │ -5.0 │
│ 3   │ 3 │ 4 │ 7 │ -1 │ -7.0 │
```
"""
add_to(q::AbstractQuery, selections...) =
    basic_select(q, query_names(q)..., selections...)

get_names(q::AbstractQuery) = query_names(q)
get_names(any) = Set{Symbol}()

# there MUST be a better way to do this...
simple_join(f, q1::AbstractQuery, q2; into = Symbol[], default_if_empty = false) = begin
    q1_keys = query_names(q1)
    q2_keys =
        if isempty(into)
            query_names(q2)
        else
            Set{Symbol}()
        end
    q2_names = get_names(q2)

    join_columns = intersect(q1_keys, q2_names)

    anon_name_1 = gensym()
    anon_name_2 = gensym()

    all_dict = immutable_merge(
        add_dots(q1_keys, anon_name_1),
        add_dots(setdiff(q2_keys, q1_keys), anon_name_2))

    selection_expression = make_selection(setdiff(union(q1_keys, q2_keys), into)...)

    if !isempty(into)
        push!(selection_expression.args, :($(into[1]) = $anon_name_2))
    end

    anons = double_with(:(($anon_name_1, $anon_name_2) -> $selection_expression), all_dict)

    if isa(q2, AbstractQuery)
        if isempty(join_columns)
            eval(:($cross_join($q1, $q2, $(anons...))))
        else
            first_join = collect(join_columns)[1]
            eval(:($f($q1, $q2,
                $(with_anonymous(first_join, q1_keys, source_name = anon_name_1)...),
                $(with_anonymous(first_join, q2_names, source_name = anon_name_2)...),
                $(anons...))))
        end
    else
        q2_expr =
            if default_if_empty
                :($(Query.default_if_empty)($q2))
            else
                q2
            end

        eval(:($f($q1,
            $(with_anonymous(:($(Query.query)($q2_expr)), q1_keys)...),
            $(anons...))))
    end
end

inner_join(q1::AbstractQuery, q2::AbstractQuery) =
    simple_join(Query.join, q1, q2)

grouped_join(q1::AbstractQuery, q2::AbstractQuery, into) =
    simple_join(Query.group_join, q1, q2; into = [get_expression(into)])

"""
    @split_rows q1::AbstractQuery *by* column

```jldoctest
julia> using LazyQuery

julia> @new_environment;

julia> @chain @evaluate begin
            using LazyQuery
            Dict(:a => [1, 2 ,3], :b => [4, 5])
            query(it)
            @make_from it first = row[1] second = row[2]
            @split_rows it by second
            @output it into DataFrame
        end
5×2 DataFrames.DataFrame
│ Row │ first │ second │
├─────┼───────┼────────┤
│ 1   │ a     │ 1      │
│ 2   │ a     │ 2      │
│ 3   │ a     │ 3      │
│ 4   │ b     │ 4      │
│ 5   │ b     │ 5      │
```
"""
split_rows(q1::AbstractQuery, by_symbol, column; default_if_empty = false) = begin
    raw_column = get_expression(column)
    simple_join(query_split_rows, q1, raw_column; into = [raw_column], default_if_empty = default_if_empty)
end

left_join(q1::AbstractQuery, q2::AbstractQuery) = begin
    tempvar = gensym()
    q1_names = query_names(q1)
    extras = [q1_names..., map(setdiff(query_names(q2), q1_names)) do value
        :($tempvar.$value)
    end...]
    make_from(split_rows(grouped_join(q1, q2, tempvar), :by, tempvar, default_if_empty = true), extras...)
end

"""
    fuze q1::AbstractQuery *conjunction* q2::AbstractQuery

```jldoctest
julia> using LazyQuery

julia> @new_environment;

julia> @chain @evaluate begin
            using LazyQuery
            df1 = DataFrame(a = [1, 2, 3], b = [4, 5, 6])
            df2 = DataFrame(a = [1, 2, 4], c = [7, 8, 9])
            @fuze query(df1) and query(df2)
            @output it into DataFrame
        end
2×3 DataFrames.DataFrame
│ Row │ a │ b │ c │
├─────┼───┼───┼───┤
│ 1   │ 1 │ 4 │ 7 │
│ 2   │ 2 │ 5 │ 8 │

julia> @chain @evaluate begin
            @fuze query(df2) into query(df1)
            @output it into DataFrame
        end
3×3 DataFrames.DataFrame
│ Row │ a │ b │ c  │
├─────┼───┼───┼────┤
│ 1   │ 1 │ 4 │ 7  │
│ 2   │ 2 │ 5 │ 8  │
│ 3   │ 3 │ 6 │ NA │

julia> @chain @evaluate begin
            df3 = DataFrame(c = [1, 2])
            @fuze query(df1) and query(df3)
            @output it into DataFrame
        end
6×3 DataFrames.DataFrame
│ Row │ a │ b │ c │
├─────┼───┼───┼───┤
│ 1   │ 1 │ 4 │ 1 │
│ 2   │ 1 │ 4 │ 2 │
│ 3   │ 2 │ 5 │ 1 │
│ 4   │ 2 │ 5 │ 2 │
│ 5   │ 3 │ 6 │ 1 │
│ 6   │ 3 │ 6 │ 2 │

julia> @evaluate @fuze query(df1) or query(df2)
ERROR: This type of join is not implemented for queries
[...]
```
"""
fuze(q1::AbstractQuery, conjunction, q2::AbstractQuery) = begin
    word = get_expression(conjunction)
    if word == :into
        left_join(q2, q1)
    elseif word == :and
        inner_join(q1, q2)
    else
        error("This type of join is not implemented for queries")
    end
end

output(q::AbstractQuery, conjunction, into) =
    eval(get_module(into), :($collect($q, $(evaluate(into)))))

output(q::AbstractQuery) = collect(q)
